package com.starnet.yarnmonitor.yarn.utils;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.starnet.yarnmonitor.yarn.dto.KafkaType;
import com.starnet.yarnmonitor.yarn.dto.KafkaWarn;
import com.starnet.yarnmonitor.yarn.entity.YarnWarnHistory;
import com.starnet.yarnmonitor.yarn.mapper.YarnWarnHistoryMapper;
import lombok.extern.log4j.Log4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;
import java.time.LocalDateTime;

@Log4j
@Component
public class KafkaConsumerUtil {
    @Autowired
    YarnWarnHistoryMapper yarnWarnHistoryMapper;

    @KafkaListener(groupId = "1", topics = "yarn-warn")
    public void listen(String msg) {
        KafkaWarn kafkaWarn = JSONObject.parseObject(msg, KafkaWarn.class);
        String kafkaType = kafkaWarn.getKafkaType().getType();
        YarnWarnHistory yarnWarnHistory = kafkaWarn.getYarnWarnHistory();
        if (kafkaType.equals("告警")) {
            dealWarn(yarnWarnHistory);
        }
        if (kafkaType.equals("恢复")) {
            dealRecovery(yarnWarnHistory);
        }
    }

    public void dealRecovery(YarnWarnHistory yarnWarnHistory) {
        //如果有未恢复的预警，恢复一下,如果没有未恢复的预警，就跳过这次
        LambdaQueryWrapper<YarnWarnHistory> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(YarnWarnHistory::getWarnId, yarnWarnHistory.getWarnId());
        queryWrapper.eq(YarnWarnHistory::getOptionId, yarnWarnHistory.getOptionId());
        queryWrapper.eq(YarnWarnHistory::getIsRecovered, false);
        YarnWarnHistory yarnWarnHistoryINDB = yarnWarnHistoryMapper.selectOne(queryWrapper);
        if (yarnWarnHistoryINDB == null) {
            return;
        }
        //有正在告警的，恢复
        yarnWarnHistoryINDB.setLimitValue(yarnWarnHistory.getLimitValue());
        yarnWarnHistoryINDB.setNowValue(yarnWarnHistory.getNowValue());
        yarnWarnHistoryINDB.setIsRecovered(true);
        yarnWarnHistoryINDB.setRecoveryTime(yarnWarnHistory.getRecoveryTime());

        int i = yarnWarnHistoryMapper.updateById(yarnWarnHistoryINDB);
        if (i != 1) {
            log.error("自动恢复告警失败,告警:" + yarnWarnHistoryINDB);
        } else {
            log.info("自动恢复告警成功，告警内容：" + yarnWarnHistoryINDB);
        }
    }


    public void dealWarn(YarnWarnHistory yarnWarnHistory) {
        //查看数据库中有没有未恢复的
        LambdaQueryWrapper<YarnWarnHistory> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(YarnWarnHistory::getWarnId, yarnWarnHistory.getWarnId());
        queryWrapper.eq(YarnWarnHistory::getOptionId, yarnWarnHistory.getOptionId());
        queryWrapper.eq(YarnWarnHistory::getIsRecovered, false);
        YarnWarnHistory yarnWarnHistoryINDB = yarnWarnHistoryMapper.selectOne(queryWrapper);
        //没有就创建新告警
        if (yarnWarnHistoryINDB == null) {
            int insert = yarnWarnHistoryMapper.insert(yarnWarnHistory);
            log.info("新增告警" + yarnWarnHistory);
            if (insert != 1) {
                log.info("异常写入数据库失败");
            }
        }
        //已有告警，更新心跳
        if (yarnWarnHistoryINDB != null) {
            yarnWarnHistoryINDB.setLastWarnTime(yarnWarnHistory.getLastWarnTime());
            yarnWarnHistoryINDB.setNowValue(yarnWarnHistory.getNowValue());
            yarnWarnHistoryINDB.setLimitValue(yarnWarnHistory.getLimitValue());
            yarnWarnHistoryMapper.updateById(yarnWarnHistoryINDB);
        }
    }

}



